package study.mq.kafka.consumer.listener._2_offset;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.stereotype.Component;

/**
 * （★）手动提交位移
 * （1）enable-auto-commit=false
 * （2）ack-mode=MANUAL_IMMEDIATE
 * （3）使用 Consumer.commitSync 方法提交位移
 */
@Slf4j
@Component
public class _2_CommitListener {

//    @KafkaListener(id = "OFFSET", topics = "_2_offset-topic")
    public void onMessage(ConsumerRecord<String, String> record, Consumer consumer) {
        //
        String topic = record.topic();
        int partition = record.partition();
        String key = record.key();
        long offset = record.offset();

        log.info(">>>>> [{}, {}] current_offset= {}", topic, partition, offset);

        /*
         *（★）同步提交
         * （1）一种阻塞式方法，方法会根据 poll() 方法拉取的最新位移来进行提交
         * （2）只要没有发生不可回复的错误，它就会阻塞消费者线程直至位移提交完成
         */
//        consumer.commitSync();
//        consumer.commitSync(Duration.ofMillis(500) );
//        consumer.commitSync(Map< TopicPartition, OffsetAndMetadata > offsets);
//        consumer.commitSync(Map<TopicPartition, OffsetAndMetadata> offsets, Duration timeout);
    }
}
